package com.study.flinktable;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.OldCsv;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

/**
 * @author sjw
 * @className TableStream
 * @date 2020/12/1 10:06
 * @description: TODO
 */
public class TableStreamWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv=StreamTableEnvironment.create(env);

        String path=TableStreamWordCount.class.getClassLoader().getResource("words.txt").getPath();

        tableEnv.connect(new FileSystem().path(path))
                .withFormat(new OldCsv()
                        .field("name", Types.STRING)
                        .field("age",Types.INT)
                        .fieldDelimiter(",").lineDelimiter(";"))
                .withSchema(new Schema()
                        .field("name", Types.STRING)
                        .field("age",Types.INT))
                .inAppendMode()
                .registerTableSource("user_info");

//        Table result=tableEnv.scan("user_info").where("age>1");
                //.select("name,age");

        Table result=tableEnv.sqlQuery("select * from user_info where age=2");

        tableEnv.toRetractStream(result, Row.class).print();

        env.execute("table");



    }

}
